package com.lz.flow.usage;

import java.util.Arrays;
import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.SubmissionPublisher;

/**
 * @Classname: com.lz.flow.usage.MySubscriber
 * @Description: TODO
 *
 * @Date: 2019/1/8 10:54
 * @Created by: li
 * @Version: V1.0
 */
public class MySubscriber<T> implements Subscriber<T> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("onSubscribe Method");
        this.subscription = subscription;
        this.subscription.request(100); //这里要使用Long.MAX_VALUE就会被认为获取无穷的数据。
    }

    @Override
    public void onNext(T item) {
        System.out.println("onNext Method");
        System.out.println("Got : " + item);
        subscription.request(110); //这里也可以使用Long.MAX_VALUE
    }

    @Override
    public void onError(Throwable t) {
        t.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("Done");
    }



    public static void main(String[] args) throws InterruptedException {

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //Register Subscriber
        MySubscriber<String> subscriber = new MySubscriber<>();
        publisher.subscribe(subscriber);
        //Publish items
        System.out.println("Publishing Items...");
        String[] items = {"1", "x", "2", "x", "3", "x"};
        Arrays.asList(items).stream().forEach(i -> publisher.submit(i));
        Thread.sleep(1000);
        Arrays.asList(items).stream().forEach(i -> publisher.submit(i));
        Thread.sleep(1000);
        publisher.close();
    }
}

